package day2

import Utils.SparkUtils
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD

import java.sql.{Driver, DriverManager, ResultSet}

object Test2 {
  def main(args: Array[String]): Unit = {
    val sc: SparkContext = SparkUtils.getSparkContext(4, "ceshi2")

    // rdd1 有10个元素 2个分区
    /*val rdd1: RDD[Int] = sc.makeRDD(1 to 10, 2)

    val array: Array[Int] = rdd1.map(v => v * 2).collect()
    array.foreach(v => println(v))*/

    /*val arr = rdd1.mapPartitions(iter => {
      val iter2: Iterator[Int] = iter.map(v => v * 2)
      iter2
    }).collect()
    arr.foreach(v => println(v))*/

    val rdd1: RDD[(Int, Int, Int)] = sc.makeRDD(List((1001, 1, 5), (1002, 2, 1), (1001, 1, 4)), 2)
    rdd1.map(v => {
      val conn = DriverManager.getConnection("jdbc:mysql://hadoop10:3306/test_db?useSSL=false&useUnicode=true&characterEncoding=utf8",
        "root",
        "123456")
      val ps = conn.prepareStatement("select * from t_product where pid = ?")
      ps.setInt(1, v._2)
      val rs: ResultSet = ps.executeQuery()
      var name: String = null
      var price: Int = 0
      if (rs.next()) {
        name = rs.getString("pname")
        price = rs.getInt("price")
      }
      conn.close()
      (v._1, v._2, name, v._3 * price)
    }).foreach(v => println(v))


    //-----------报错的话就把关闭资源去掉

    /* val rdd1: RDD[(Int, Int, Int)] = sc.makeRDD(List((1001, 1, 5), (1002, 2, 1), (1001, 1, 4)), 2)
     rdd1.mapPartitions(iter => {
       val conn = DriverManager.getConnection("jdbc:mysql://hadoop10:3306/test_db?useSSL=false&useUnicode=true&characterEncoding=utf8",
         "root",
         "123456")
       val ps = conn.prepareStatement("select * from t_product where pid = ?")
       val iter2 = iter.map(v => {
         ps.setInt(1, v._2)
         val rs = ps.executeQuery()
         var name: String = null
         var price: Int = 0
         if (rs.next()) {
           name = rs.getString("pname")
           price = rs.getInt("price")
         }
         (v._1, v._2, name, v._3 * price)

       }).foreach(v => println(v))
       iter2
     })*/
  }
}
